package com.rk.topN;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class TopNReducer extends Reducer<ViewCountBean, Text, Text, ViewCountBean> {

    private TreeMap<ViewCountBean, Text> map = new TreeMap<ViewCountBean, Text>();

    @Override
    protected void reduce(ViewCountBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for(Text value : values){
            ViewCountBean bean = new ViewCountBean();
            bean.setDocId(key.getDocId());
            bean.setViewTotalCount(key.getViewTotalCount());

            map.put(bean, new Text(value));

            if(map.size() > 10){
                map.remove(map.lastKey());
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        Iterator<ViewCountBean> it = map.keySet().iterator();
        while (it.hasNext()){
            ViewCountBean bean = it.next();
            context.write(new Text(map.get(bean)), bean);
        }
    }
}
